package com.tech.common

import com.tech.config.ApplicationConfig
import com.tech.process.nodeProcess._
import org.apache.kudu.client.KuduClient.KuduClientBuilder
import org.apache.kudu.client._

import java.time.LocalDate
import java.util.UUID
import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object KuduUtil {

  @transient lazy val kuduClient: KuduClient = new KuduClientBuilder(ApplicationConfig.KUDU_MASTER).build()
  @transient lazy val session: KuduSession = kuduClient.newSession()
  //事件表
  @transient lazy val retaileventTable: KuduTable = kuduClient.openTable("retailevent")
  //用户群组表
  @transient lazy val customerGroupDetailTable: KuduTable = kuduClient.openTable("customer_group_detail")
  //用户活动群组表
  @transient lazy val customerGroupDetail4ProcessTable: KuduTable = kuduClient.openTable("customer_group_detail_for_process")
  //用户参与活动记录表
  @transient lazy val customerActivityTable: KuduTable = kuduClient.openTable("customer_activity")
  //用户行为节点记录表
  @transient lazy val customerBehaviorNodeTable: KuduTable = kuduClient.openTable("customer_node_behavior")
  //用户通过节点记录表
  @transient lazy val customerNodeTable: KuduTable = kuduClient.openTable("customer_node")
  //用户表
  @transient lazy val customerTable: KuduTable = kuduClient.openTable("customer")
  //延时节点表
  @transient lazy val delayNodeTable: KuduTable = kuduClient.openTable("delay_node")
  //行为属性标签群组表
  @transient lazy val dimDataGroupDetailTable: KuduTable = kuduClient.openTable("dimdatagroupdetail")
  //列属性表
  @transient lazy val columnTypeTable: KuduTable = kuduClient.openTable("column_type")
  //行为详情记录表
  @transient lazy val customerBehaviorRecordTable: KuduTable = kuduClient.openTable("customer_behavior_record")
  //offset表
  @transient lazy val offsetTable: KuduTable = kuduClient.openTable("offset")

  //===========================
  //以下表为D365专用的活动元数据表
  //===========================

  //活动详情表
  @transient lazy val tsr_marketing_activity: KuduTable = kuduClient.openTable("tsr_marketing_activity")
  //节点表
  @transient lazy val tsr_marketing_activity_process: KuduTable = kuduClient.openTable("tsr_marketing_activity_process")
  //客户节点详情表
  @transient lazy val tsr_marketing_process_customer: KuduTable = kuduClient.openTable("tsr_marketing_process_customer")
  //行为节点详情表
  @transient lazy val tsr_marketing_process_behavior: KuduTable = kuduClient.openTable("tsr_marketing_process_behavior")
  //A/B Test节点详情表
  @transient lazy val tsr_marketing_process_test: KuduTable = kuduClient.openTable("tsr_marketing_process_test")
  //等待节点详情表
  @transient lazy val tsr_marketing_process_wait: KuduTable = kuduClient.openTable("tsr_marketing_process_wait")


  /**
   * 检查客户是否参与当前活动
   *
   * @param customer_id 用户ID
   * @param activity_id 活动ID
   * @return Boolean
   */
  def check(customer_id: String, activity_id: String): Boolean = {

    // 设置查询条件
    val schema = customerActivityTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)
    val activityId = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_id"), KuduPredicate.ComparisonOp.EQUAL, activity_id)

    // 执行查询操作
    val scanner = kuduClient.newScannerBuilder(customerActivityTable)
      .setProjectedColumnNames(List("customer_id", "activity_id").asJava)
      .addPredicate(projectPredicate)
      .addPredicate(customerId)
      .addPredicate(activityId)
      .build()

    val result = scanner.nextRows().hasNext
    scanner.close()

    result

  }

  /**
   * 插入需要延时处理的消息
   *
   * @param customer_id  用户id
   * @param activity_id  活动id
   * @param process_id   节点id
   * @param action_time  触发时间
   * @param node_type    节点类型
   * @param event_key    事件类型
   * @param limit_name   限制字段
   * @param limit_symbol 比较条件
   * @param is_group     是否是客群
   * @param group_id     客群id
   * @param group_table  客群来源表
   *
   */
  def insertDelayNode(project: String,
                      customer_id: String,
                      member_id: String,
                      activity_id: String,
                      process_id: String,
                      action_time: Long,
                      node_type: String,
                      event_key: String = null,
                      current_value: Double = -1,
                      limit_value: Double = -1,
                      limit_value_min: Double = -1,
                      limit_value_max: Double = -1,
                      limit_symbol: String = null,
                      is_group: String,
                      group_id: String = null): Unit = {

    val upsert = delayNodeTable.newUpsert()
    val row = upsert.getRow

    row.addString("project", ApplicationConfig.PROJECT)
    row.addString("customer_id", customer_id)
    row.addString("member_id", member_id)
    row.addString("activity_id", activity_id)
    row.addString("process_id", process_id)
    row.addLong("action_time", action_time * 1000)
    row.addString("node_type", node_type)
    if (event_key != null)
      row.addString("event_key", event_key)
    if (current_value != -1)
      row.addDouble("current_value", current_value)
    if (limit_value != -1)
      row.addDouble("limit_value", limit_value)
    if (limit_value_min != -1)
      row.addDouble("limit_value_min", limit_value_min)
    if (limit_value_max != -1)
      row.addDouble("limit_value_max", limit_value_max)
    if (limit_symbol != null)
      row.addString("limit_symbol", limit_symbol)
    row.addString("is_group", is_group)
    if (group_id != null)
      row.addString("group_id", group_id)
    row.addLong("create_time", System.currentTimeMillis() * 1000)

    session.apply(upsert)
    session.flush()
  }

  /**
   * 客群初始化时批量插入需要延时处理的行为节点消息
   *
   * @param customer_id  用户id
   * @param activity_id  活动id
   * @param process_id   节点id
   * @param action_time  触发时间
   * @param node_type    节点类型
   * @param event_key    事件类型
   * @param limit_name   限制字段
   * @param limit_symbol 比较条件
   * @param is_group     是否是客群
   * @param group_id     客群id
   * @param group_table  客群来源表
   *
   */
  def insertDelayNode4Map(group_id: String,
                          project: String,
                          activity_id: String,
                          process_id: String,
                          action_time: Long,
                          node_type: String,
                          event_key: String = null,
                          current_value: Double = -1,
                          limit_value: Double = -1,
                          limit_value_min: Double = -1,
                          limit_value_max: Double = -1,
                          limit_symbol: String = null,
                          is_group: String = "0"): Unit = {

    val set = getCustomerIdAndMemberIdByGroupId(group_id)

    set.foreach(
      i => {
        insertDelayNode(ApplicationConfig.PROJECT, i._1, i._2, activity_id, process_id, action_time, node_type, event_key, current_value, limit_value, limit_value_min, limit_value_max, limit_symbol, is_group, group_id)
      }
    )

  }

  /**
   * 插入客户群组
   *
   * @param map （groupId -> Set(customerId)）
   */
  def insertGroups4Map(map: mutable.HashMap[String, mutable.Set[String]], project: String, activity_id: String): Unit = {

    map.foreach(
      i => {
        var num = 1
        i._2.foreach(
          customerId => {
            val upsert = customerGroupDetail4ProcessTable.newUpsert()
            val row = upsert.getRow
            row.addString("group_id", i._1)
            row.addString("cust_id", customerId)
            val member_id = getMemberId(customerId)
            if (member_id != null)
              row.addString("member_id", customerId)
            row.addString("project", ApplicationConfig.PROJECT)
            row.addString("activity_id", activity_id)
            row.addString("created_time", SimplyDateUtil.getFormat().format(System.currentTimeMillis()))
            row.addInt("number", num)
            session.apply(upsert)
            session.flush()
            num += 1
          }
        )
      }
    )


  }


  /**
   * 插入数据到kudu
   *
   * @param tableName 表名
   * @param map       插入的map
   * @param schemaMap 表的schema
   */
  def insertRow(tableName: String, map: mutable.HashMap[String, Any], schemaMap: mutable.HashMap[String, String]): Unit = {

    val table = tableName match {
      case "retailevent" =>
        retaileventTable
      case "customer" =>
        customerTable
    }

    val upsert = table.newUpsert()
    val row = upsert.getRow

    map.foreach(
      i => {
        if (schemaMap.contains(i._1)) {
          if (i._2 != null)
            schemaMap(i._1) match {
              case "string" =>
                row.addString(i._1, i._2.toString)
              case "unixtime_micros" =>
                row.addLong(i._1, i._2.toString.toLong * 1000)
              case "double" =>
                row.addDouble(i._1, i._2.toString.toDouble)
              case "int32" =>
                row.addInt(i._1, i._2.toString.toInt)
            }
        }
      }
    )

    session.apply(upsert)
    session.flush()


  }

  /**
   * 向用户参与活动记录表中插入数据
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   */
  def insertCustomerActivity(customer_id: String, activity_id: String): Unit = {

    val nodeUpsert = customerActivityTable.newUpsert()
    nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
    nodeUpsert.getRow.addString("customer_id", customer_id)
    nodeUpsert.getRow.addString("activity_id", activity_id)
    nodeUpsert.getRow.addLong("create_time", System.currentTimeMillis() * 1000)

    session.apply(nodeUpsert)
    session.flush()

  }

  /**
   * 向用户参与活动记录表中批量插入群组数据
   *
   * @param group_id    群组id
   * @param activity_id 活动id
   */
  def insertCustomerActivity4Group(group_id: String, activity_id: String): Unit = {

    val set = getIdsByGroupId(group_id)

    set.foreach(
      customer_id => {
        val nodeUpsert = customerActivityTable.newUpsert()
        nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
        nodeUpsert.getRow.addString("customer_id", customer_id)
        nodeUpsert.getRow.addString("activity_id", activity_id)
        nodeUpsert.getRow.addLong("create_time", System.currentTimeMillis() * 1000)

        session.apply(nodeUpsert)
        session.flush()
      }
    )

  }

  /**
   * 在用户通过节点记录表中插入数据
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param process_id  节点id
   */
  def insertPassNodeMsg(customer_id: String, activity_id: String, process_id: String): Unit = {

    val nodeUpsert = customerNodeTable.newUpsert()
    nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
    nodeUpsert.getRow.addString("customer_id", customer_id)
    nodeUpsert.getRow.addString("activity_id", activity_id)
    nodeUpsert.getRow.addString("process_id", process_id)
    nodeUpsert.getRow.addLong("create_time", System.currentTimeMillis() * 1000)

    session.apply(nodeUpsert)
    session.flush()

  }

  /**
   * 在用户通过节点记录表中批量插入群组数据
   *
   * @param group_id    群组id
   * @param activity_id 活动id
   * @param process_id  节点id
   */
  def insertPassNodeMsg4Group(group_id: String, activity_id: String, process_id: String): Unit = {

    val set = getCustomerIdAndMemberIdByGroupId(group_id)

    set.foreach(
      i => {
        val nodeUpsert = customerNodeTable.newUpsert()
        nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
        nodeUpsert.getRow.addString("customer_id", i._1)
        nodeUpsert.getRow.addString("activity_id", activity_id)
        nodeUpsert.getRow.addString("process_id", process_id)
        nodeUpsert.getRow.addLong("create_time", System.currentTimeMillis() * 1000)

        //将消息逐条发送到kafka 2021/11/1 不发送group_id相关消息，改为按customer_id逐条发送
        //        KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, ApplicationConfig.PROJECT, activity_id, process_id, i._1, i._2)
        insertPassNodeMsg(i._1, activity_id, process_id)

        session.apply(nodeUpsert)
        session.flush()
      }
    )

  }


  /**
   * 在用户节点表上插入新用户的节点信息(仅保留行为节点)
   *
   * @param customer_id 用户id
   * @param startNode   用户节点信息
   */
  def insertStartNode(customer_id: String, startNode: NextNode, event_key: String): Unit = {

    val nodeUpsert = customerBehaviorNodeTable.newUpsert()
    nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
    nodeUpsert.getRow.addString("customer_id", customer_id)
    nodeUpsert.getRow.addString("activity_id", startNode.activity_id)
    nodeUpsert.getRow.addString("process_id", startNode.id)
    nodeUpsert.getRow.addString("event_key", event_key)
    nodeUpsert.getRow.addLong("create_time", System.currentTimeMillis() * 1000)

    session.apply(nodeUpsert)
    session.flush()

  }


  /**
   * 向用户行为详情记录表上插入数据
   *
   * @param event_key        事件类型
   * @param activity_id      活动id
   * @param customer_id      顾客id
   * @param process_id       节点id
   * @param member_id        会员id
   * @param source_channel   来源渠道
   * @param happened_channel 发生渠道
   * @param event_time       事件时间
   */
  def insertBehaviorRecord(event_key: String,
                           activity_id: String,
                           customer_id: String,
                           process_id: String,
                           member_id: String,
                           source_channel: Int,
                           happened_channel: String,
                           event_time: Long): Unit = {

    val nodeUpsert = customerBehaviorRecordTable.newUpsert()

    nodeUpsert.getRow.addString("id", UUID.randomUUID().toString)
    nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
    nodeUpsert.getRow.addString("behavior_name", ApplicationConfig.EVENT_MAP.getOrElse(event_key, event_key))
    nodeUpsert.getRow.addString("activity_id", activity_id)
    nodeUpsert.getRow.addString("customer_id", customer_id)
    nodeUpsert.getRow.addString("process_id", process_id)
    nodeUpsert.getRow.addString("member_id", member_id)
    if (source_channel != -1)
      nodeUpsert.getRow.addInt("source_channel", source_channel)
    if (happened_channel != null)
      nodeUpsert.getRow.addString("happened_channel", happened_channel)
    nodeUpsert.getRow.addLong("happend_time", event_time * 1000)
    nodeUpsert.getRow.addLong("create_time", System.currentTimeMillis() * 1000)

    session.apply(nodeUpsert)
    session.flush()

  }

  /**
   * 在用户节点表上更新用户的节点信息(仅保留行为节点)
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param process_id  节点id
   * @param event_key   事件key
   * @param is_delete   软删除  0:否  1:是
   * @param create_time 创建时间
   */
  def upsertNode(customer_id: String,
                 activity_id: String,
                 process_id: String,
                 event_key: String,
                 is_delete: String,
                 create_time: Long,
                 limit_name: String = null,
                 limit_symbol: String = null,
                 limit_value: Double = -1,
                 limit_value_min: Double = -1,
                 limit_value_max: Double = -1,
                 event_value: Double = -1,
                 event_num: Double = -1): Unit = {

    val nodeUpsert = customerBehaviorNodeTable.newUpsert()
    println(customer_id, activity_id, process_id, event_key, is_delete)
    nodeUpsert.getRow.addString("project", ApplicationConfig.PROJECT)
    nodeUpsert.getRow.addString("customer_id", customer_id)
    nodeUpsert.getRow.addString("activity_id", activity_id)
    nodeUpsert.getRow.addString("process_id", process_id)
    nodeUpsert.getRow.addString("event_key", event_key)
    if (is_delete != null)
      nodeUpsert.getRow.addString("is_delete", is_delete)
    nodeUpsert.getRow.addLong("create_time", create_time * 1000)
    if (limit_name != null)
      nodeUpsert.getRow.addString("limit_name", limit_name)
    if (limit_symbol != null)
      nodeUpsert.getRow.addString("limit_symbol", limit_symbol)
    if (limit_value != -1)
      nodeUpsert.getRow.addDouble("limit_value", limit_value)
    if (limit_value_min != -1)
      nodeUpsert.getRow.addDouble("limit_value_min", limit_value_min)
    if (limit_value_max != -1)
      nodeUpsert.getRow.addDouble("limit_value_max", limit_value_max)
    if (event_value != -1)
      nodeUpsert.getRow.addDouble("event_value", event_value)
    if (event_num != -1)
      nodeUpsert.getRow.addDouble("event_num", event_num)

    session.apply(nodeUpsert)
    session.flush()

  }

  /**
   * 客群活动初始化时在用户节点表上更新用户的节点信息(仅保留行为节点)
   *
   * @param group_id    客群id
   * @param activity_id 活动id
   * @param process_id  节点id
   * @param event_key   事件key
   * @param is_delete   软删除  0:否  1:是
   * @param create_time 创建时间
   */
  def upsertNode4Map(group_id: String,
                     activity_id: String,
                     process_id: String,
                     event_key: String,
                     is_delete: String,
                     create_time: Long,
                     limit_name: String = null,
                     limit_symbol: String = null,
                     limit_value: Double = -1,
                     limit_value_min: Double = -1,
                     limit_value_max: Double = -1,
                     event_value: Double = -1,
                     event_num: Double = -1
                    ): Unit = {

    val set = getIdsByGroupId(group_id)

    set.foreach(
      customer_id => {
        upsertNode(customer_id, activity_id, process_id, event_key, is_delete, create_time, limit_name, limit_symbol, limit_value, limit_value_min, limit_value_max, event_value, event_num)
      }
    )


  }


  /**
   * 获取表schema
   *
   * @param tableName 表名
   * @return Map(字段名 -> 字段类型)
   */
  def getTableSchema(tableName: String): mutable.HashMap[String, String] = {

    val table = tableName match {
      case "retailevent" =>
        retaileventTable
      case "customer" =>
        customerTable
    }

    val schema = table.getSchema

    val iterator = schema.getColumns.iterator()

    val map = mutable.HashMap[String, String]()

    while (iterator.hasNext) {

      val columnSchema = iterator.next()
      map.put(columnSchema.getName, columnSchema.getType.getName)

    }

    map

  }

  /**
   * 获取客户群组中的客户id集
   *
   * @param group_id 群组ID
   * @return 客户群组的客户id集
   */
  def getIdsByGroupId(group_id: String): mutable.Set[String] = {

    val set: mutable.Set[String] = mutable.Set()

    // 设置查询条件
    //    val schema = customerGroupDetail4ProcessTable.getSchema
    val schema = customerGroupDetailTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val groupId = KuduPredicate.newComparisonPredicate(schema.getColumn("group_id"), KuduPredicate.ComparisonOp.EQUAL, group_id)

    // 执行查询操作
    //    val scanner = kuduClient.newScannerBuilder(customerGroupDetail4ProcessTable)
    val scanner = kuduClient.newScannerBuilder(customerGroupDetailTable)
      .setProjectedColumnNames(List("group_id", "cust_id").asJava)
      .addPredicate(projectPredicate)
      .addPredicate(groupId)
      .build()
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        set.add(result.getString("cust_id"))
      }
    }
    scanner.close()

    set

  }

  /**
   * 获取客户群组中的(customer_id,member_id)集
   *
   * @param group_id 群组ID
   * @return 客户群组的(customer_id,member_id)集
   */
  def getCustomerIdAndMemberIdByGroupId(group_id: String): mutable.Set[(String, String)] = {

    val set: mutable.Set[(String, String)] = mutable.Set()

    // 设置查询条件
    val schema = customerGroupDetail4ProcessTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val groupId = KuduPredicate.newComparisonPredicate(schema.getColumn("group_id"), KuduPredicate.ComparisonOp.EQUAL, group_id)

    // 执行查询操作
    val scanner = kuduClient.newScannerBuilder(customerGroupDetail4ProcessTable)
      .setProjectedColumnNames(List("group_id", "cust_id", "member_id").asJava)
      .addPredicate(projectPredicate)
      .addPredicate(groupId)
      .build()
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        if (!result.isNull("cust_id") && !result.isNull("member_id"))
          set.add(result.getString("cust_id"), result.getString("member_id"))
      }
    }
    scanner.close()

    set

  }


  /**
   * 获取列对应属性
   *
   * @return Map(column -> type)
   */
  def getColumnTypeMap: mutable.HashMap[String, String] = {

    // 设置查询条件
    val schema = customerGroupDetail4ProcessTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)

    val scanner = kuduClient.newScannerBuilder(columnTypeTable)
      .addPredicate(projectPredicate)
      .build()

    val map = new mutable.HashMap[String, String]()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val column = result.getString("column")
        val _type = result.getString("type")
        map.put(column, _type)
      }
    }

    map
  }

  /**
   * 查询当前用户所在活动节点中可以与当前事件进行规则匹配的节点
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param event_key   事件类型
   * @return 需要进行实时规则匹配的节点Array
   */
  def getNodes(customer_id: String, activity_id: String, event_key: String): ArrayBuffer[String] = {

    // 设置查询条件
    val schema = customerBehaviorNodeTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)
    val activityId = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_id"), KuduPredicate.ComparisonOp.EQUAL, activity_id)
    val eventKey = KuduPredicate.newComparisonPredicate(schema.getColumn("event_key"), KuduPredicate.ComparisonOp.EQUAL, event_key)
    val isDelete = KuduPredicate.newIsNullPredicate(schema.getColumn("is_delete"))

    val array = new ArrayBuffer[String]
    // 执行查询操作
    val builder = kuduClient.newScannerBuilder(customerBehaviorNodeTable).setProjectedColumnNames(List("process_id").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(customerId)
    builder.addPredicate(activityId)
    builder.addPredicate(eventKey)
    builder.addPredicate(isDelete)

    val scanner = builder.build()

    //    println(println("开始循环 ==> " + event_key + "    " + customer_id + "     " + System.currentTimeMillis()))
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val process_id = result.getString("process_id")
        array.append(process_id)
      }
    }
    //    println(println("结束循环 ==> " + event_key + "    " + customer_id + "     " + System.currentTimeMillis()))

    scanner.close()

    array

  }

  /**
   * 获取用户当前节点累计满足的total
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param process_id  节点id
   * @return eventValue(Double)
   */
  def getNodeEventValue(customer_id: String, activity_id: String, process_id: String): Double = {

    // 设置查询条件
    val schema = customerBehaviorNodeTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)
    val activityId = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_id"), KuduPredicate.ComparisonOp.EQUAL, activity_id)
    val processId = KuduPredicate.newComparisonPredicate(schema.getColumn("process_id"), KuduPredicate.ComparisonOp.EQUAL, process_id)
    val isDelete = KuduPredicate.newIsNullPredicate(schema.getColumn("is_delete"))
    // 执行查询操作
    val builder = kuduClient.newScannerBuilder(customerBehaviorNodeTable).setProjectedColumnNames(List("event_value").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(customerId)
    builder.addPredicate(activityId)
    builder.addPredicate(processId)
    builder.addPredicate(isDelete)

    val scanner = builder.build()

    var eventValue = 0D
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        if (!result.getDouble("event_value").isNaN)
          eventValue = result.getDouble("event_value")
      }

    }
    scanner.close()

    eventValue

  }


  /**
   * 获取用户当前节点累计满足的total
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param process_id  节点id
   * @return eventValue(Double)
   */
  def getNodeRow(customer_id: String, activity_id: String, process_id: String): RowResult = {

    // 设置查询条件
    val schema = customerBehaviorNodeTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)
    val activityId = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_id"), KuduPredicate.ComparisonOp.EQUAL, activity_id)
    val processId = KuduPredicate.newComparisonPredicate(schema.getColumn("process_id"), KuduPredicate.ComparisonOp.EQUAL, process_id)
    val isDelete = KuduPredicate.newIsNullPredicate(schema.getColumn("is_delete"))
    // 执行查询操作
    val builder = kuduClient.newScannerBuilder(customerBehaviorNodeTable)
    //      .setProjectedColumnNames(List("event_value").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(customerId)
    builder.addPredicate(activityId)
    builder.addPredicate(processId)
    builder.addPredicate(isDelete)

    val scanner = builder.build()

    var rowResult: RowResult = null

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        rowResult = iterator.next()
      }

    }
    scanner.close()

    rowResult

  }


  /**
   * 获取用户首次进入当前节点的时间（即上个节点结束的时间）
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param process_id  节点id
   * @return 首次进入当前节点的时间
   */
  def getNodeTime(customer_id: String, activity_id: String, process_id: String): Long = {

    // 设置查询条件
    val schema = customerBehaviorNodeTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val customerId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)
    val activityId = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_id"), KuduPredicate.ComparisonOp.EQUAL, activity_id)
    val processId = KuduPredicate.newComparisonPredicate(schema.getColumn("process_id"), KuduPredicate.ComparisonOp.EQUAL, process_id)
    val isDelete = KuduPredicate.newIsNullPredicate(schema.getColumn("is_delete"))
    // 执行查询操作
    val builder = kuduClient.newScannerBuilder(customerBehaviorNodeTable).setProjectedColumnNames(List("create_time").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(customerId)
    builder.addPredicate(activityId)
    builder.addPredicate(processId)
    builder.addPredicate(isDelete)

    val scanner = builder.build()

    var create_time = 0L
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        create_time = result.getLong("create_time")
      }

    }
    scanner.close()

    create_time / 1000

  }

  /**
   * 获取用户详细信息
   *
   * @param customer_id 用户id
   * @return RowResult
   */
  def getCustomerDetail(customer_id: String): RowResult = {

    // 设置查询条件
    val schema = customerTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val groupId = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)

    // 执行查询操作
    val scanner = kuduClient.newScannerBuilder(customerTable)
      .addPredicate(projectPredicate)
      .addPredicate(groupId)
      .build()
    var result: RowResult = null
    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        result = iterator.next()
      }
    }
    scanner.close()

    result

  }

  /**
   * 获取行为属性标签群组
   *
   * @return map(群组id -> array(标签值))
   */
  def getDimDataGroupDetail: mutable.HashMap[String, ArrayBuffer[String]] = {
    println("仅初始化一次")
    // 设置查询条件
    val schema = customerTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    // 执行查询操作
    val scanner = kuduClient.newScannerBuilder(dimDataGroupDetailTable)
      .addPredicate(projectPredicate)
      .build()
    val map: mutable.HashMap[String, ArrayBuffer[String]] = mutable.HashMap("" -> ArrayBuffer(""))

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val key = result.getString("group_id")
        val value = result.getString("key")
        if (map != null && map.contains(key)) {
          map(key).append(value)
        } else {
          map.put(key, ArrayBuffer[String](value))
        }
      }
    }
    scanner.close()

    map

  }

  def getMemberId(customer_id: String): String = {

    // 设置查询条件
    val schema = customerTable.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val customerPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("customer_id"), KuduPredicate.ComparisonOp.EQUAL, customer_id)
    val scanner = kuduClient.newScannerBuilder(customerTable)
      .addPredicate(projectPredicate)
      .addPredicate(customerPredicate)
      .build()

    var result: String = null
    if (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      if (iterator.hasNext) {
        val rowResult = iterator.next()
        if (!rowResult.isNull("member_id")) {
          result = rowResult.getString("member_id")
        }
      }
    }
    result

  }


  /**
   * 获取上次消费截止的offset+1
   *
   * @return Map(topic-partition -> offset)
   */
  def getOffset(): mutable.HashMap[String, Long] = {

    val builder = kuduClient.newScannerBuilder(offsetTable)

    val scanner = builder.build()

    val map = new mutable.HashMap[String, Long]()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val topic = result.getString("topic")
        val partition = result.getInt("partition")
        val offset = result.getLong("offset")
        map.put(topic + "-" + partition, offset)
      }
    }

    map
  }

  /**
   * 更新消费到的offset
   */
  def upsertOffset(topic: String, partition: Int, offset: Long): Unit = {

    val upsert = offsetTable.newUpsert()
    val row = upsert.getRow

    row.addString("topic", topic)
    row.addInt("partition", partition)
    row.addLong("offset", offset + 1)
    row.addLong("create_time", System.currentTimeMillis() * 1000)

    session.apply(upsert)
    session.flush()

  }


  //=========================
  //以下接口为适应D365环境专门适配
  //=========================

  /**
   * 获取活跃活动
   *
   * @return 活跃活动ID
   */
  def getActivity4D365: ArrayBuffer[String] = {

    val now = LocalDate.now()
    val array = new ArrayBuffer[String]()

    // 设置查询条件
    val schema = tsr_marketing_activity.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val statusCondition = KuduPredicate.newComparisonPredicate(schema.getColumn("status"), KuduPredicate.ComparisonOp.EQUAL, 100000002)
    val beginTimeCondition = KuduPredicate.newComparisonPredicate(schema.getColumn("begin_time"), KuduPredicate.ComparisonOp.LESS_EQUAL, System.currentTimeMillis() * 1000)
    val endTimeCondition = KuduPredicate.newComparisonPredicate(schema.getColumn("end_time"), KuduPredicate.ComparisonOp.GREATER_EQUAL, System.currentTimeMillis() * 1000)
    val activityTypeEq0 = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_type"), KuduPredicate.ComparisonOp.EQUAL, 0)
    val activityTypeEq1 = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_type"), KuduPredicate.ComparisonOp.EQUAL, 1)
    val periodUnitEq100000000 = KuduPredicate.newComparisonPredicate(schema.getColumn("period_unit"), KuduPredicate.ComparisonOp.EQUAL, 100000000)
    val periodUnitEq100000001 = KuduPredicate.newComparisonPredicate(schema.getColumn("period_unit"), KuduPredicate.ComparisonOp.EQUAL, 100000001)
    val periodUnitEq100000002 = KuduPredicate.newComparisonPredicate(schema.getColumn("period_unit"), KuduPredicate.ComparisonOp.EQUAL, 100000002)
    val periodUnitEq100000003 = KuduPredicate.newComparisonPredicate(schema.getColumn("period_unit"), KuduPredicate.ComparisonOp.EQUAL, 100000003)
    val periodValueDayOfYear = KuduPredicate.newComparisonPredicate(schema.getColumn("period_value"), KuduPredicate.ComparisonOp.EQUAL, now.getDayOfYear)
    val periodValueDayOfMonth = KuduPredicate.newComparisonPredicate(schema.getColumn("period_value"), KuduPredicate.ComparisonOp.EQUAL, now.getDayOfMonth)
    val periodValueDayOfWeek = KuduPredicate.newComparisonPredicate(schema.getColumn("period_value"), KuduPredicate.ComparisonOp.EQUAL, now.getDayOfWeek.getValue)

    //筛选活跃活动（非周期）
    val builder = kuduClient.newScannerBuilder(tsr_marketing_activity)
      .setProjectedColumnNames(List("id").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(statusCondition)
    builder.addPredicate(beginTimeCondition)
    builder.addPredicate(endTimeCondition)
    builder.addPredicate(activityTypeEq0)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        array.append(result.getString("id"))
      }

    }
    scanner.close()

    //筛选周期性活动（年）
    val builder1 = kuduClient.newScannerBuilder(tsr_marketing_activity).setProjectedColumnNames(List("id").asJava)
    builder1.addPredicate(projectPredicate)
    builder1.addPredicate(statusCondition)
    builder1.addPredicate(beginTimeCondition)
    builder1.addPredicate(endTimeCondition)
    builder1.addPredicate(activityTypeEq1)
    builder1.addPredicate(periodUnitEq100000000)
    builder1.addPredicate(periodValueDayOfYear)

    val scanner1 = builder1.build()

    while (scanner1.hasMoreRows) {
      val iterator = scanner1.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        array.append(result.getString("id"))
      }

    }
    scanner1.close()

    //筛选周期性活动（月）
    val builder2 = kuduClient.newScannerBuilder(tsr_marketing_activity).setProjectedColumnNames(List("id").asJava)
    builder2.addPredicate(projectPredicate)
    builder2.addPredicate(statusCondition)
    builder2.addPredicate(beginTimeCondition)
    builder2.addPredicate(endTimeCondition)
    builder2.addPredicate(activityTypeEq1)
    builder2.addPredicate(periodUnitEq100000001)
    builder2.addPredicate(periodValueDayOfMonth)

    val scanner2 = builder2.build()

    while (scanner2.hasMoreRows) {
      val iterator = scanner2.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        array.append(result.getString("id"))
      }

    }
    scanner2.close()


    //筛选周期性活动（周）
    val builder3 = kuduClient.newScannerBuilder(tsr_marketing_activity).setProjectedColumnNames(List("id").asJava)
    builder3.addPredicate(projectPredicate)
    builder3.addPredicate(statusCondition)
    builder3.addPredicate(beginTimeCondition)
    builder3.addPredicate(endTimeCondition)
    builder3.addPredicate(activityTypeEq1)
    builder3.addPredicate(periodUnitEq100000002)
    builder3.addPredicate(periodValueDayOfWeek)

    val scanner3 = builder3.build()

    while (scanner3.hasMoreRows) {
      val iterator = scanner3.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        array.append(result.getString("id"))
      }

    }
    scanner3.close()

    //筛选周期性活动（日）
    val builder4 = kuduClient.newScannerBuilder(tsr_marketing_activity).setProjectedColumnNames(List("id").asJava)
    builder4.addPredicate(projectPredicate)
    builder4.addPredicate(statusCondition)
    builder4.addPredicate(beginTimeCondition)
    builder4.addPredicate(endTimeCondition)
    builder4.addPredicate(activityTypeEq1)
    builder4.addPredicate(periodUnitEq100000003)

    val scanner4 = builder4.build()

    while (scanner4.hasMoreRows) {
      val iterator = scanner4.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        array.append(result.getString("id"))
      }

    }
    scanner4.close()

    array

  }


  /**
   * 获取静态客户节点
   *
   * @param activity_id 活动id
   * @return map（节点id -> 客群id）
   */
  def getGroupNodes4D365(activity_id: String): mutable.HashMap[String, String] = {

    val map = mutable.HashMap[String, String]()
    // 设置查询条件
    val schema = tsr_marketing_process_customer.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newComparisonPredicate(schema.getColumn("activity_id"), KuduPredicate.ComparisonOp.EQUAL, activity_id)
    val notDynamicCondition = KuduPredicate.newComparisonPredicate(schema.getColumn("group_type"), KuduPredicate.ComparisonOp.EQUAL, 100000000)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_customer).setProjectedColumnNames(List("process_id", "customer_group_id").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)
    builder.addPredicate(notDynamicCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        map.put(result.getString("process_id"), result.getString("customer_group_id"))
      }

    }
    scanner.close()

    map
  }


  /**
   * 获取行为节点及其对应事件类型
   *
   * @param activities 活跃活动
   * @return Map（节点id -> 事件类型）
   */
  def getBehaviorMap4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, String] = {

    val map = mutable.HashMap[String, String]()
    // 设置查询条件
    val schema = tsr_marketing_process_behavior.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_behavior).setProjectedColumnNames(List("process_id", "event_type").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        map.put(result.getString("process_id"), result.getString("event_type"))
      }

    }
    scanner.close()

    map

  }


  /**
   * 获取开始节点
   *
   * @param activities 活跃活动
   * @return Map（活动id -> 开始节点id）
   */
  def getStartNodes4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, String] = {

    val map = mutable.HashMap[String, String]()
    // 设置查询条件
    val schema = tsr_marketing_activity_process.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)
    val prevNodeCondition = KuduPredicate.newIsNullPredicate(schema.getColumn("prev_node"))

    val builder = kuduClient.newScannerBuilder(tsr_marketing_activity_process).setProjectedColumnNames(List("activity_id", "id").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)
    builder.addPredicate(prevNodeCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        map.put(result.getString("activity_id"), result.getString("id"))
      }

    }
    scanner.close()

    map

  }


  /**
   * 获取活动节点、下级节点
   *
   * @param activities 活跃活动
   * @return Map（节点id -> 下级节点信息（NextNode））
   */
  def getNextNodes4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, NextNode] = {

    val map = new mutable.HashMap[String, NextNode]()
    // 设置查询条件
    val schema = tsr_marketing_activity_process.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_activity_process).setProjectedColumnNames(List("activity_id", "id", "type", "next_node").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val node = new NextNode(
          result.getString("activity_id"),
          result.getString("id"),
          result.getLong("type").toInt,
          if (!result.isNull("next_node"))
            result.getString("next_node")
          else
            null
        )
        map.put(result.getString("id"), node)
      }

    }
    scanner.close()

    map
  }


  /**
   * 获取客群
   *
   * @param activities 活跃活动
   * @return Map（客户节点id -> Set(客户id)）
   */
  def getGroups4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, mutable.Set[String]] = {

    val map = new mutable.HashMap[String, mutable.Set[String]]
    // 设置查询条件
    val schema = tsr_marketing_process_customer.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_customer).setProjectedColumnNames(List("process_id", "customer_group_id").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        if (!map.keySet.contains(result.getString("process_id")))
          map.put(result.getString("customer_group_id"), getIdsByGroupId(result.getString("customer_group_id")))
      }

    }
    scanner.close()

    map

  }


  /**
   * 获取全部行为节点规则
   *
   * @param activities 活跃活动
   * @return Map（行为节点id -> 行为节点规则(BehaviorNode)）
   */
  def getBehaviorNodes4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, BehaviorNode] = {

    val map = new mutable.HashMap[String, BehaviorNode]()
    // 设置查询条件
    val schema = tsr_marketing_process_behavior.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_behavior)
      .setProjectedColumnNames(List("activity_id", "process_id", "time_type", "time_condition", "customer_behavior", "event_type", "event_name", "addition", "measure_property", "behavior_attribute").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val node = new BehaviorNode(
          result.getString("activity_id"),
          result.getString("process_id"),
          result.getLong("time_type").toInt,
          result.getString("time_condition"),
          result.getLong("customer_behavior").toInt,
          result.getString("event_type"),
          result.getString("event_name"),
          if (!result.isNull("addition"))
            result.getString("addition")
          else
            null
          ,
          result.getString("measure_property"),
          result.getString("behavior_attribute")
        )
        map.put(result.getString("process_id"), node)
      }

    }
    scanner.close()

    map

  }

  /**
   * 获取全部行为节点类型
   *
   * @param activities 活跃活动
   * @return Map（行为节点id -> 行为节点类型（event_key））
   */
  def getEventKeys4D365(activities: ArrayBuffer[String]): mutable.HashSet[String] = {

    val set = new mutable.HashSet[String]()
    // 设置查询条件
    val schema = tsr_marketing_process_behavior.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_behavior)
      .setProjectedColumnNames(List("event_type").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        set.add(result.getString("event_type"))
      }

    }
    scanner.close()

    set

  }


  /**
   * 获取全部客户节点规则
   *
   * @param activities 活跃活动
   * @return Map（客户节点id -> 客户节点规则(CustomerNode)）
   */
  def getCustomerNodes4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, CustomerNode] = {

    val map = new mutable.HashMap[String, CustomerNode]()
    // 设置查询条件
    val schema = tsr_marketing_process_customer.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_customer)
      .setProjectedColumnNames(List("activity_id", "process_id", "customer_group_id", "customer_group_name").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val node = new CustomerNode(
          result.getString("activity_id"),
          result.getString("process_id"),
          result.getString("customer_group_id"),
          result.getString("customer_group_name")
        )
        map.put(result.getString("process_id"), node)
      }

    }
    scanner.close()

    map

  }


  /**
   * 获取全部 A/B TEST 节点规则
   *
   * @param activities 活跃活动
   * @return Map（A/B TEST节点id -> A/B TEST节点规则(TestNode)）
   */
  def getTestNodes4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, TestNode] = {

    val map = new mutable.HashMap[String, TestNode]()
    // 设置查询条件
    val schema = tsr_marketing_process_test.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_test)
      .setProjectedColumnNames(List("activity_id", "process_id", "probability").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val node = new TestNode(
          result.getString("activity_id"),
          result.getString("process_id"),
          result.getLong("probability").toInt
        )
        map.put(result.getString("process_id"), node)
      }

    }
    scanner.close()

    map

  }


  /**
   * 获取全部等待节点规则
   *
   * @param activities 活跃活动
   * @return Map（等待节点id -> 等待节点规则(WaitNode)）
   */
  def getWaitNodes4D365(activities: ArrayBuffer[String]): mutable.HashMap[String, WaitNode] = {

    val map = new mutable.HashMap[String, WaitNode]()
    // 设置查询条件
    val schema = tsr_marketing_process_wait.getSchema
    val projectPredicate = KuduPredicate.newComparisonPredicate(schema.getColumn("project"), KuduPredicate.ComparisonOp.EQUAL, ApplicationConfig.PROJECT)
    val activityCondition = KuduPredicate.newInListPredicate(schema.getColumn("activity_id"), activities.asJava)

    val builder = kuduClient.newScannerBuilder(tsr_marketing_process_wait)
      .setProjectedColumnNames(List("activity_id", "process_id", "type", "wait_duration", "wait_date", "unit").asJava)
    builder.addPredicate(projectPredicate)
    builder.addPredicate(activityCondition)

    val scanner = builder.build()

    while (scanner.hasMoreRows) {
      val iterator = scanner.nextRows()
      while (iterator.hasNext) {
        val result = iterator.next()
        val node = new WaitNode(
          result.getString("activity_id"),
          result.getString("process_id"),
          result.getLong("type").toInt,
          if (!result.isNull("wait_duration"))
            result.getLong("wait_duration").toInt
          else -1,
          if (!result.isNull("wait_date"))
            result.getLong("wait_date") / 1000
          else -1,
          if (!result.isNull("unit"))
            result.getLong("unit").toInt
          else -1
        )
        map.put(result.getString("process_id"), node)
      }

    }
    scanner.close()

    map

  }


}
